package cn.jly.flink.state;

import cn.jly.flink.entity.SensorReading;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 传感器连续10s温度持续上升报警
 * 方案：
 * 第一条数据来的时候，注册一个10s的定时器，并收集这10s的数据。
 * 当定时器触发的时候，判断温度是否持续上升。
 * 如果状态中是空，则把当前温度放进去；
 * 新的温度来的时候，如果是上升，则更新状态值；如果下降，则清空状态值，并删除定时器
 *
 * @PackageName cn.jly.flink.state
 * @ClassName PracticeDemo2
 * @Description 传感器连续10s温度持续上升报警
 * @Author 姬岚洋
 * @Date 2021/2/4 下午 4:58
 */
public class PracticeDemo2 {
    private static final String CONTINUE_INCREMENT_TEMP_TAG = "continueIncrementTemp";

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();

        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        final String host = parameterTool.get("host", "192.168.172.101");
        final int port = parameterTool.getInt("port", 9999);

        // outputTag
        final OutputTag<Tuple2<String, SensorReading>> continueIncrementTempTag = new OutputTag<>(
                CONTINUE_INCREMENT_TEMP_TAG,
                TypeInformation.of(new TypeHint<Tuple2<String, SensorReading>>() {
                })
        );

        final SingleOutputStreamOperator<Tuple2<String, SensorReading>> processDataStream = env.socketTextStream(host, port)
                .flatMap(new FlatMapFunction<String, SensorReading>() {
                    @Override
                    public void flatMap(String value, Collector<SensorReading> out) throws Exception {
                        if (StringUtils.isNotEmpty(value)) {
                            final String[] split = value.split(",");
                            out.collect(new SensorReading(split[0], Long.parseLong(split[1]), Double.parseDouble(split[2])));
                        }
                    }
                })
                .keyBy(new KeySelector<SensorReading, String>() {
                    @Override
                    public String getKey(SensorReading value) throws Exception {
                        return value.getId();
                    }
                })
                .process(new KeyedProcessFunction<String, SensorReading, Tuple2<String, SensorReading>>() {
                    // 上一次的传感器信息状态
                    transient ValueState<SensorReading> lastSensorReadingState = null;
                    // 定时器状态
                    transient ValueState<Long> timerTsState = null;

                    final Integer interval = 10;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        final ValueStateDescriptor<SensorReading> lastSensorReadingValueStateDescriptor = new ValueStateDescriptor<>(
                                "lastSensorReadingValueStateDescriptor",
                                TypeInformation.of(new TypeHint<SensorReading>() {
                                })
                        );
                        final ValueStateDescriptor<Long> timerTsDescriptor = new ValueStateDescriptor<>(
                                "timerTsDescriptor",
                                Long.class
                        );

                        // 初始化状态
                        lastSensorReadingState = getRuntimeContext().getState(lastSensorReadingValueStateDescriptor);
                        timerTsState = getRuntimeContext().getState(timerTsDescriptor);
                    }

                    @Override
                    public void processElement(SensorReading currentSensorReading, Context ctx, Collector<Tuple2<String, SensorReading>> out) throws Exception {
                        // 取出状态数据
                        final SensorReading lastSensorReading = lastSensorReadingState.value();
                        final Long timerTs = timerTsState.value();
                        // 判断是否上升
                        if (lastSensorReading != null) {
                            // 如果温度上升，并且没有定时器的时候，注册一个10s定时器，然后等待
                            if (currentSensorReading.getTemperature() > lastSensorReading.getTemperature() && timerTs == null) {
                                // 计算出定时器时间戳
                                long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;
                                // 注册定时器
                                ctx.timerService().registerProcessingTimeTimer(ts);
                                // 更新状态
                                timerTsState.update(timerTs);
                            } else if (currentSensorReading.getTemperature() <= lastSensorReading.getTemperature() && timerTs != null) {
                                // 如果温度下降，且定时器不为空，则删除定时器
                                ctx.timerService().deleteProcessingTimeTimer(timerTs);
                                // 清空定时器状态
                                timerTsState.clear();
                            }
                        }

                        // 更新温度状态
                        lastSensorReadingState.update(currentSensorReading);

                        // 时刻打印温度
                        out.collect(Tuple2.of("当前温度", currentSensorReading));
                    }

                    /**
                     * 定时器触发
                     *
                     * @param timestamp
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, SensorReading>> out) throws Exception {
                        ctx.output(continueIncrementTempTag, Tuple2.of("传感器" + ctx.getCurrentKey() + "温度连续" + interval + "s持续上升", lastSensorReadingState.value()));
                        // 触发定时器之后清空定时器
                        timerTsState.clear();
                    }
                });

        // 时刻输出温度
        processDataStream.print("current");

        // 打印报警信息
        processDataStream.getSideOutput(continueIncrementTempTag).printToErr("warning");

        FlinkUtils.executeStream(env, "PracticeDemo2");
    }
}
